package com.zjl.学习.第08章_配置文件.test;

import com.zjl.学习.第07章_高级.User;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@SpringBootApplication
@EnableBatchProcessing
public class C0_分区步骤 {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    //设置  每个分区文件读取
    @Bean//任务 Tasklet 这是最简单的    还有Chunk块处理
    public Tasklet tasklet001(){
        return new Tasklet() {
            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                Map<String, Object> stepExecutionContext = chunkContext.getStepContext().getStepExecutionContext();

                System.out.println(stepExecutionContext);
                return RepeatStatus.FINISHED;//已完成
            }
        };
    }


    //文件分区器-设置分区规则
    @Bean
    public MyPartitioner myPartitioner(){
        return new MyPartitioner();
    }

    //文件分区处理器-处理分区  设置分区信息
    @Bean
    public PartitionHandler userPartitionHandler() {
        TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
        handler.setGridSize(2);//表示有5个分区
        handler.setTaskExecutor(new SimpleAsyncTaskExecutor());//表示各分区异步处理
        handler.setStep(workStep());//放入分步骤的  步骤
        try {
            handler.afterPropertiesSet();//进行 步骤判空
        } catch (Exception e) {
            e.printStackTrace();
        }
        return handler;
    }

    //每个从分区操作步骤   配置分区的 shep 步骤
    @Bean
    public Step workStep() {
        TaskletStep workStep =  stepBuilderFactory.get("workStep")
                .tasklet(tasklet001())
                .build();
        return workStep;
    }

    //主分区操作步骤
    @Bean
    public Step masterStep() {
        return stepBuilderFactory.get("masterStep")
                //放入分步骤的 名字，表示对谁进行步骤的分区操作    放入分区规则
                .partitioner(workStep().getName(), myPartitioner())
                .partitionHandler(userPartitionHandler())//放入分区器
                .build();
    }


    @Bean
    public Job partJob(){
        return jobBuilderFactory.get("part-job")
                .start(masterStep())

                .incrementer(new RunIdIncrementer())

                .build();
    }
    public static void main(String[] args) {
        System.setProperty("spring.profiles.active","xml");
        SpringApplication.run(C0_分区步骤.class, args);
        System.out.println("***********启动完毕**当前编码:" + Charset.defaultCharset().name() + "****************");

    }
}class MyPartitioner implements Partitioner {

    @Override//gridSize  表示你是第几个运行的分区
    public Map<String, ExecutionContext> partition(int gridSize) {
        Map<String, ExecutionContext> result = new HashMap<>(gridSize);

        int range = 10; //文件间隔
        int start = 1; //开始位置
        int end = 10;  //结束位置
        String text = "item/fq/user%s-%s.txt";

        for (int i = 0; i < 5; i++) {
            ExecutionContext value = new ExecutionContext();//上下文，就是某一个分区的指定数据
            String format = String.format(text, start, end);
            Resource resource = new ClassPathResource(format);//为每个分区放入数据
            try {
                value.putString("file", resource.getURL().toExternalForm());
                value.putString("fileName", format);
            } catch (IOException e) {
                e.printStackTrace();
            }
            start += range;
            end += range;

            result.put("user_partition_" + i, value);
        }
        return result;
    }
}

